package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object Demo18Cache {

  def main(args: Array[String]): Unit = {


    //环境配置对象
    val conf = new SparkConf()

    //指定任务名
    conf.setAppName("pi")
    conf.setMaster("local")


    //创建spark 环境,式写spark代码的入口
    val sc = new SparkContext(conf)


    val studentRDD: RDD[String] = sc.textFile("data/students.txt")

    val kvRDD: RDD[(String, Int)] = studentRDD.map(line => {
      println("map")

      val split: Array[String] = line.split(",")

      (split(4), split(2).toInt)
    })

    /**
      * 如果同一个rdd被多次使用，可以将rdd的数据加载到内存中
      *
      * 缓存级别选择
      * 1、如果数据量没有超过内存的限制使用MEMORY_ONLY
      * 2、如果数据量比较大，MEMORY_AND_DISK_SER，尽量将数据压到内存中
      *    将数据缓存到Executor的内存或者磁盘上
      *
      */

    //默认放内存中
    //kvRDD.cache()

    kvRDD.persist(StorageLevel.MEMORY_AND_DISK_SER)


    //统计班级的人数
    kvRDD
      .map(kv => (kv._1, 1))
      .reduceByKey(_ + _)
      .foreach(println)


    //统计班级的平均年龄
    kvRDD
      .groupByKey()
      .map(kv => {
        val clazz: String = kv._1
        val ages: List[Int] = kv._2.toList
        val avgAge: Double = ages.sum.toDouble / ages.size
        (clazz, avgAge)
      })
      .foreach(println)


    while (true) {

    }

  }

}
